package beautiful.butterfly.drds.data_exchange.core.job.scheduler.processinner;

import beautiful.butterfly.drds.data_exchange.core.job.scheduler.AbstractScheduler;
import beautiful.butterfly.drds.data_exchange.core.statistics.container.communicator.AbstractContainerCommunicator;
import beautiful.butterfly.drds.data_exchange.core.taskgroup.TaskGroupContainer;
import beautiful.butterfly.drds.data_exchange.core.taskgroup.runner.TaskGroupContainerRunner;
import beautiful.butterfly.drds.data_exchange.core.util.FrameworkErrorCode;
import common.exception.DataExchangeException;
import common.util.Configuration;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public abstract class ProcessInnerScheduler extends AbstractScheduler
{

    private ExecutorService taskGroupContainerExecutorService;

    public ProcessInnerScheduler(AbstractContainerCommunicator containerCommunicator)
    {
        super(containerCommunicator);
    }

    @Override
    public void startAllTaskGroup(List<Configuration> configurationList)
    {
        this.taskGroupContainerExecutorService = Executors
                .newFixedThreadPool(configurationList.size());

        for (Configuration taskGroupConfiguration : configurationList)
        {
            TaskGroupContainerRunner taskGroupContainerRunner = newTaskGroupContainerRunner(taskGroupConfiguration);
            this.taskGroupContainerExecutorService.execute(taskGroupContainerRunner);
        }

        this.taskGroupContainerExecutorService.shutdown();
    }

    @Override
    public void dealFailedStat(AbstractContainerCommunicator frameworkCollector, Throwable throwable)
    {
        this.taskGroupContainerExecutorService.shutdownNow();
        throw DataExchangeException.asDataExchangeException(
                FrameworkErrorCode.PLUGIN_RUNTIME_ERROR, throwable);
    }


    @Override
    public void dealKillingStat(AbstractContainerCommunicator frameworkCollector, int totalTasks)
    {
        //通过进程退出返回码标示状态
        this.taskGroupContainerExecutorService.shutdownNow();
        throw DataExchangeException.asDataExchangeException(FrameworkErrorCode.KILLED_EXIT_VALUE,
                "job killed status");
    }


    private TaskGroupContainerRunner newTaskGroupContainerRunner(
            Configuration configuration)
    {
        TaskGroupContainer taskGroupContainer = new TaskGroupContainer(configuration);

        return new TaskGroupContainerRunner(taskGroupContainer);
    }

}
